package xj.toolkit.bsh;

import bsh.Interpreter;
import xj.toolkit.util.ExceptionUtils;

import java.io.FilterReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintStream;
import java.io.Reader;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import javax.management.Notification;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jmx.export.notification.NotificationPublisher;
import org.springframework.jmx.export.notification.NotificationPublisherAware;

public class BshConsole implements NotificationPublisherAware {
	private static final Logger LOG = LoggerFactory.getLogger(BshConsole.class);

	public void setNotificationPublisher(
			NotificationPublisher notificationPublisher) {
		this._notificationPublisher = notificationPublisher;
	}

	public BshConsole() {
		this._runbsh = initSingleExecutor("bsh executor");
		this._runInterpreterFuture.set(createAndRunBshInterpreter());

		this._runpublish = initSingleExecutor("bsh output publish executor");
		startAsyncPublish();
	}

	public String getVersion() {
		return "0.0.1";
	}

	private Future<?> createAndRunBshInterpreter() {
		try {
			PrintStream out = new PrintStream(new ByteArrayQueueOutputStream(
					this._outputByteArrays));
			final Interpreter interpreter = new Interpreter(

			new InputStreamReader(new ByteQueueInputStream(this._inputBytes),
					"UTF-8"),

			out, out, true);
			interpreter.setExitOnEOF(false);

			this._interpreterRef.set(interpreter);

			return this._runbsh.submit(new Runnable() {
				public void run() {
					interpreter.run();
				}
			});
		} catch (UnsupportedEncodingException e) {
			LOG.error("exception when create Interpreter, detail: {}",
					ExceptionUtils.exception2detail(e));
			throw new RuntimeException(e);
		}
	}

	private void startAsyncPublish() {
		this._runpublish.submit(new Runnable() {
			public void run() {
				try {
					byte[] bytes = (byte[]) BshConsole.this._outputByteArrays
							.poll(1000L, TimeUnit.MILLISECONDS);
					if (bytes != null) {
						BshConsole.this._notificationPublisher
								.sendNotification(new Notification(
										"bsh.output", "UnitAdmin",
										BshConsole.this._notifSequence++,
										new String(bytes)));
					} else if (BshConsole.LOG.isTraceEnabled()) {
						BshConsole.LOG
								.trace("async print has !NO! content, goto next round.");
					}
				} catch (InterruptedException localInterruptedException) {
					if (BshConsole.LOG.isInfoEnabled()) {
						BshConsole.LOG.info("async print fetch interrupted.");
					}
				} finally {
					BshConsole.this.startAsyncPublish();
				}
			}
		});
	}

	public void stop() {
		if (this._runbsh != null) {
			this._runbsh.shutdownNow();
		}
		if (this._runpublish != null) {
			this._runpublish.shutdownNow();
		}
	}

	public void resetBsh() {
		Future<?> current = (Future) this._runInterpreterFuture.getAndSet(null);
		if (current != null) {
			current.cancel(true);
			this._runInterpreterFuture.set(createAndRunBshInterpreter());
		} else {
			LOG.warn("other resetInterpreter IN-PROGRESS, just ignore this call");
		}
	}

	public void inputIntoBsh(String input) {
		try {
			byte[] bytes = input.getBytes("UTF-8");
			List<Byte> Bytes = new ArrayList(bytes.length);
			for (byte b : bytes) {
				Bytes.add(Byte.valueOf(b));
			}
			Bytes.add(new Byte((byte) -1));
			this._inputBytes.addAll(Bytes);
		} catch (UnsupportedEncodingException e) {
			LOG.error("exception when getBytes, detail: {}",
					ExceptionUtils.exception2detail(e));
		}
	}

	public String execute(String[] vars, String[] scripts,
			boolean useSeparateInterpreter) {
		Interpreter interpreter = gainInterpreter(useSeparateInterpreter);
		if (vars != null) {
			for (int idx = 0; idx < vars.length / 2 * 2; idx += 2) {
				try {
					interpreter.set(vars[idx], vars[(idx + 1)]);
				} catch (Exception e) {
					LOG.error(
							"exception when set var {} with value {}, detail: {}",
							new Object[] { vars[idx], vars[(idx + 1)],
									ExceptionUtils.exception2detail(e) });
				}
			}
		}
		Object retval = null;
		if (scripts != null) {
			for (int idx = 0; idx < scripts.length; idx++) {
				try {
					retval = interpreter.eval(scripts[idx]);
				} catch (Exception e) {
					LOG.error(
							"exception when execute [{}] scripts, detail: {}",
							Integer.valueOf(idx),
							ExceptionUtils.exception2detail(e));
				}
			}
		}
		return retval != null ? retval.toString() : null;
	}

	private Interpreter gainInterpreter(boolean useSeparateInterpreter) {
		if (useSeparateInterpreter) {
			return new Interpreter();
		}
		Interpreter interpreter = (Interpreter) this._interpreterRef.get();
		return interpreter != null ? interpreter : new Interpreter();
	}

	private static ExecutorService initSingleExecutor(String threadName) {
		ExecutorService executor = Executors.newFixedThreadPool(1,
				new ThreadFactory() {
					public Thread newThread(Runnable r) {
						return new Thread(r);
					}
				});
		Future<?> future = executor.submit(new Runnable() {
			public void run() {
				if (BshConsole.LOG.isDebugEnabled()) {
					BshConsole.LOG.debug("Init thread call");
				}
			}
		});
		try {
			future.get();
		} catch (Exception e) {
			LOG.error("exception when init thread named {}, detail: {}",
					threadName, ExceptionUtils.exception2detail(e));
		}
		return executor;
	}

	private final AtomicReference<Future<?>> _runInterpreterFuture = new AtomicReference(
			null);
	private final AtomicReference<Interpreter> _interpreterRef = new AtomicReference(
			null);
	private final BlockingQueue<Byte> _inputBytes = new LinkedBlockingQueue();
	private final BlockingQueue<byte[]> _outputByteArrays = new LinkedBlockingQueue();
	private final ExecutorService _runbsh;
	private final ExecutorService _runpublish;
	private NotificationPublisher _notificationPublisher;
	private long _notifSequence = 0L;

	private static class ByteQueueInputStream extends InputStream {
		private final BlockingQueue<Byte> _bytes;

		public ByteQueueInputStream(BlockingQueue<Byte> bytes) {
			this._bytes = bytes;
		}

		public int read() throws IOException {
			try {
				if (BshConsole.LOG.isTraceEnabled()) {
					BshConsole.LOG.trace("try read int value");
				}
				int i = ((Byte) this._bytes.take()).intValue();
				if (BshConsole.LOG.isTraceEnabled()) {
					BshConsole.LOG.trace("read int value: {}",
							Integer.valueOf(i));
				}
				return i;
			} catch (InterruptedException e) {
				BshConsole.LOG.warn(
						"exception when ByteQueueInputStream.read, detail: {}",
						ExceptionUtils.exception2detail(e));
				throw new RuntimeException(e);
			}
		}
	}

	private static class CommandLineReader extends FilterReader {
		static final int normal = 0;
		static final int lastCharNL = 1;
		static final int sentSemi = 2;

		public CommandLineReader(Reader in) {
			super(in);
		}

		int state = 1;

		public int read() throws IOException {
			if (this.state == 2) {
				this.state = 1;
				return 10;
			}
			int b;
			while ((b = this.in.read()) == 13) {
			}
			if (b == 10) {
				if (this.state == 1) {
					b = 59;
					this.state = 2;
				} else {
					this.state = 1;
				}
			} else {
				this.state = 0;
			}
			return b;
		}

		public int read(char[] buff, int off, int len) throws IOException {
			int b = read();
			if (b == -1) {
				return -1;
			}
			buff[off] = ((char) b);
			return 1;
		}
	}

	private static class ByteArrayQueueOutputStream extends OutputStream {
		private final BlockingQueue<byte[]> _byteArrays;

		public ByteArrayQueueOutputStream(BlockingQueue<byte[]> byteArrays) {
			this._byteArrays = byteArrays;
		}

		public void write(int b) throws IOException {
			this._byteArrays.offer(new byte[] { (byte) b });
			if (BshConsole.LOG.isTraceEnabled()) {
				BshConsole.LOG.trace("output, write byte: {}",
						Integer.valueOf(b));
			}
		}

		public void write(byte[] b, int off, int len) throws IOException {
			this._byteArrays.offer(Arrays.copyOfRange(b, off, off + len));
			if (BshConsole.LOG.isTraceEnabled()) {
				BshConsole.LOG.trace("output, write bytes : {} off {} len {}",
						new Object[] { Arrays.toString(b),
								Integer.valueOf(off), Integer.valueOf(len) });
			}
		}

		public void write(byte[] b) throws IOException {
			this._byteArrays.offer(Arrays.copyOf(b, b.length));
			if (BshConsole.LOG.isTraceEnabled()) {
				BshConsole.LOG.trace("output, write bytes : {} ",
						Arrays.toString(b));
			}
		}
	}
}
